package anli

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object HotCategoryTop10Analysis {
  /**
   * 网站用户行为：搜索，点击，下单，支付
   * 数据字段：日期 用户id SessionID 页面ID 动作时间 搜索关键字
   * 需求:1、读取原始日志数据
   * 2、统计品类的点击数量：品类ID和点击次数
   * 3、统计品类的下单数量：品类ID，下单数量
   * 4、统计品类的支付数量：品类ID，支付数量
   * 5、将品类进行排序，并取前10名
   *    点击数量排序，下单数量排序，支付数量排序
   *    元组排序：先比较第一个，再比较第二个，再比较后面的，依此类推
   * 6、将结果采集到控制台打印出来
   * */
  def main(args: Array[String]): Unit = {
    val sparkconf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("anli2")
    val sc=new SparkContext(sparkconf)
    // 需求:1、读取原始日志数据
    val actionRDD=sc.textFile("data/user_visit_action.txt")
    //2、统计品类的点击数量：品类ID和点击次数
    val clickActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas = action.split("_")
        datas(6) != "-1"
      }
      
    )
    val clickCountRDD: RDD[(String, Int)] = clickActionRDD.map(
      action => {
        val datas = action.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_ + _).sortBy(_._2,false)
    println("统计品类的点击数量：品类ID和点击次数")
    clickCountRDD.collect().foreach(println)
    // 3、统计品类的下单数量：品类ID，下单数量
    val orderActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas = action.split("_")
        datas(8) != "null"
      }
    )
    val orderCountRDD: RDD[(String, Int)] = orderActionRDD.flatMap(
      action => {
        val datas: Array[String] = action.split("_")
        val cid: String = datas(8)
        val cids: Array[String] = cid.split(",")
        cids.map(id => (id, 1))
      }
    ).reduceByKey(_ + _)
//      .sortBy(_._2,false)
    println("统计品类的下单数量：品类ID，下单数量")
    orderCountRDD.collect().foreach(println)

    //4、统计品类的支付数量：品类ID，支付数量
    val payActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        datas(10) != "null"
      }
    )
    val payCountRDD: RDD[(String, Int)] = payActionRDD.flatMap(
      action => {
        val datas: Array[String] = action.split("_")
        val payid: String = datas(10)
        val pay: Array[String] = payid.split("_")
        pay.map(
          id => (id, 1)
        )
      }
    ).reduceByKey(_ + _)
//      .sortBy(_._2, false).take(6)
    println("统计品类的支付数量：品类ID，支付数量")
    payCountRDD.foreach(println)
    //5、将品类进行排序，并取前10名
    //   点击数量排序，下单数量排序，支付数量排序
    //   元组排序：先比较第一个，再比较第二个，再比较后面的，依此类推
    //将数据按照 (品类ID)，（点击数 下单数 支付数量）统计数据输出
    val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] =
     clickCountRDD.cogroup(orderCountRDD, payCountRDD)

    val analysisRDD: RDD[(String, (Int, Int, Int))] = cogroupRDD.mapValues {
      case (clickIter, orderIter, payIter) => {
        var clickCnt = 0
        val clickIterator: Iterator[Int] = clickIter.iterator
        if (clickIterator.hasNext) {
          clickCnt = clickIterator.next()
        }
        var orderCnt = 0
        val orderIterator: Iterator[Int] = orderIter.iterator
        if (orderIterator.hasNext) {
          orderCnt = orderIterator.next()
        }
        var payCnt = 0
        val payiterator: Iterator[Int] = payIter.iterator
        if (payiterator.hasNext) {
          payCnt = payiterator.next()
        }
        (clickCnt, orderCnt, payCnt)
      }
    }
    //将数据排序
    val resultRDD: Array[(String, (Int, Int, Int))] = analysisRDD.sortBy(_._2._2,false).take(10)
    //6、将结果采集到控制台打印出来
    println("将数据按照 (品类ID)，（点击数 下单数 支付数量）统计数据输出,默认按下单数据排序")
    resultRDD.foreach(println)



    sc.stop()
  }

}
